package net.lainiao;

import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.shared.SharedCount;
import org.apache.curator.framework.recipes.shared.SharedCountListener;
import org.apache.curator.framework.recipes.shared.SharedCountReader;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;

import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2016/11/9.
 */
public class SharedCounterExample implements SharedCountListener {
    private static final int QTY=5;
    private static final String PATH="/SharedCounterExample/count";

    public static void main(String[] args) throws Exception {
        final Random random=new Random();
        SharedCounterExample sharedCounterExample=new SharedCounterExample();
        try(TestingServer server=new TestingServer()){
            CuratorFramework client= CuratorFrameworkFactory.newClient(server.getConnectString(),new ExponentialBackoffRetry(1000,3));
            client.start();
            SharedCount baseCount=new SharedCount(client,PATH,0);
            baseCount.addListener(sharedCounterExample);
            baseCount.start();
            List<SharedCount> examples= Lists.newArrayList();
            ExecutorService service= Executors.newFixedThreadPool(QTY);
            for (int i = 0; i <QTY ; i++) {
                final SharedCount count=new SharedCount(client,PATH,0);
                examples.add(count);
                final int temCount=i;
                Callable<Void> task=new Callable<Void>() {
                    @Override
                    public Void call() throws Exception {
                        count.start();
                        Thread.sleep(random.nextInt(10000));
                        System.out.println("Increment:"+count.trySetCount(count.getVersionedValue(),count.getCount()+ random.nextInt(10)));
//                        count.setCount(temCount);
                        return null;
                    }
                };
                service.submit(task);
            }
            service.shutdown();
            service.awaitTermination(10, TimeUnit.MINUTES);
            for (int i = 0; i <QTY ; i++) {
                examples.get(i).close();
            }
            baseCount.close();
        }
    }

    @Override
    public void countHasChanged(SharedCountReader sharedCountReader, int newCount) throws Exception {
        System.out.println("Counter;s value is changed to :"+newCount);
    }

    @Override
    public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
        System.out.println("state changed:"+connectionState.toString());
    }
}

